// market_data_player
// 시세 처리
// rabbitmq에 publish
// redis에 set
// 2013. Nam
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <netinet/in.h>

#include <hiredis/hiredis.h>
#include <stdint.h>
#include <amqp.h>
#include <amqp_framing.h>

#define BUFSIZE 1024

int main(int argc, char const *argv[])
{
	// multicast variables
	int recv_fd;
	struct sockaddr_in recv_addr, tmp_addr;
	int tmp_addr_len = sizeof(tmp_addr);
	struct ip_mreq mreq;
	char msg[1024];
	char publish_msg[801];
	char *tmp = NULL;
	char *code_num = NULL;
	
	// redis
	redisContext *redis;
	redisReply *reply;

	// rabbitmq variables
	amqp_connection_state_t rm_conn;
	char rm_hostname[20];
	if(argc==2)
	{
		strcpy(rm_hostname,argv[1]);
		redis = redisConnect(argv[1], 6379);
	}
	else
	{
		strcpy(rm_hostname,"192.168.172.137");//172.16.52.134");
		redis = redisConnect("localhost", 6379);
	}

	if(redis->err)
	{
		fprintf(stderr, "redis접속이안된다고!!\n");
		exit(0);
	}

	char *rm_exchange="nam_sise";
	char *rm_exchangetype="topic";
	int rm_socket;
	//amqp_socket_t *rm_socket = NULL;
	int rm_port = 5672;

	// multicast setting
	recv_fd = socket(AF_INET, SOCK_DGRAM, 0);
	int sockopt = 1;

	recv_addr.sin_addr.s_addr = inet_addr("224.0.0.1");
	recv_addr.sin_port = htons(9999);
	recv_addr.sin_family = AF_INET;

	mreq.imr_multiaddr = recv_addr.sin_addr;
	mreq.imr_interface.s_addr = htonl(INADDR_ANY);
	setsockopt(recv_fd, SOL_SOCKET, SO_REUSEADDR, &sockopt, sizeof(sockopt));
	setsockopt(recv_fd, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mreq, sizeof(mreq));
	bind(recv_fd, (struct sockaddr*)&recv_addr, sizeof(recv_addr));

	// rabbitmq setting
	rm_conn = amqp_new_connection();
	rm_socket = amqp_open_socket(rm_hostname, rm_port);
	//rm_socket = amqp_tcp_socket_new();
	//int rm_status = amqp_socket_open(rm_socket, rm_hostname, rm_port);
	
	amqp_set_sockfd(rm_conn, rm_socket);
	//amqp_set_socket(rm_conn, rm_socket);

	amqp_login(rm_conn,"/",0,131072,0,AMQP_SASL_METHOD_PLAIN, "guest","guest");
	amqp_channel_open(rm_conn, 1);
	amqp_get_rpc_reply(rm_conn);

	amqp_exchange_declare(rm_conn, 1, amqp_cstring_bytes(rm_exchange),
		amqp_cstring_bytes(rm_exchangetype), 0, 0, amqp_empty_table);
	amqp_get_rpc_reply(rm_conn);

	printf("ready....\n");
	int recv_flag = 0;

	char rm_routing_message[256]={0};
	char rm_publish_message[256]={0};

	while(1)
	{
		memset(msg, 0, sizeof(msg));
		memset(publish_msg, 0, sizeof(publish_msg));

		recvfrom(recv_fd, &msg, sizeof(msg), 0, (struct sockaddr*)&tmp_addr, &tmp_addr_len);
		if(recv_flag==0){ printf("Processing...\n"); }
		recv_flag = 1;
		// 데이터 파싱.
		if(strncmp(msg,"A3011",5)==0)
		{
			code_num = (char*)substring(msg,5,16); // 종목코드
			//strcpy(publish_msg, msg);
			sprintf(rm_routing_message, "%s.A3011", code_num);
			amqp_basic_publish(rm_conn, 1, amqp_cstring_bytes(rm_exchange), amqp_cstring_bytes(rm_routing_message),0,0,NULL,amqp_cstring_bytes(msg));

			reply = redisCommand(redis, "SET %s %s", rm_routing_message, msg);
			freeReplyObject(reply);

			// tmp = (char*)substring(msg, 19, 27); // 전일대비 2.4
			// sprintf(rm_routing_message, "시세.%s.전일대비", code_num);
			// amqp_basic_publish(rm_conn, 1, amqp_cstring_bytes(rm_exchange), amqp_cstring_bytes(rm_routing_message),0,0,NULL,amqp_cstring_bytes(tmp));
			// free(tmp);
			
			tmp = (char*)substring(msg, 28, 36); // 체결가 1.4
			sprintf(rm_routing_message, "%s.체결가", code_num);
			reply = redisCommand(redis, "SET %s %s", rm_routing_message, tmp);
			freeReplyObject(reply);
			free(tmp);
			
			// amqp_basic_publish(rm_conn, 1, amqp_cstring_bytes(rm_exchange), amqp_cstring_bytes(rm_routing_message),0,0,NULL,amqp_cstring_bytes(tmp));
			
			// tmp = (char*)substring(msg, 49, 57); // 시가 5.1
			// sprintf(rm_routing_message, "시세.%s.시가", code_num);
			// amqp_basic_publish(rm_conn, 1, amqp_cstring_bytes(rm_exchange), amqp_cstring_bytes(rm_routing_message),0,0,NULL,amqp_cstring_bytes(tmp));
			// free(tmp);

			// tmp = (char*)substring(msg, 58, 66); // 고가 6.1
			// sprintf(rm_routing_message, "시세.%s.고가", code_num);
			// amqp_basic_publish(rm_conn, 1, amqp_cstring_bytes(rm_exchange), amqp_cstring_bytes(rm_routing_message),0,0,NULL,amqp_cstring_bytes(tmp));
			// free(tmp);

			// tmp = (char*)substring(msg, 67, 75); // 저가 7.1
			// sprintf(rm_routing_message, "시세.%s.저가", code_num);
			// amqp_basic_publish(rm_conn, 1, amqp_cstring_bytes(rm_exchange), amqp_cstring_bytes(rm_routing_message),0,0,NULL,amqp_cstring_bytes(tmp));
			// free(tmp);

			// tmp = (char*)substring(msg, 76, 87); // 누적체결수량 4.4
			// sprintf(rm_routing_message, "시세.%s.누적체결수량", code_num);
			// amqp_basic_publish(rm_conn, 1, amqp_cstring_bytes(rm_exchange), amqp_cstring_bytes(rm_routing_message),0,0,NULL,amqp_cstring_bytes(tmp));
			// free(tmp);

			// tmp = (char*)substring(msg, 108, 113); //체결시각 title
			// sprintf(rm_routing_message, "시세.%s.체결시각", code_num);
			// amqp_basic_publish(rm_conn, 1, amqp_cstring_bytes(rm_exchange), amqp_cstring_bytes(rm_routing_message),0,0,NULL,amqp_cstring_bytes(tmp));
			// free(tmp);

			free(code_num);
			// redis set, rabbitmq publish
		}//A3011
		else if(strncmp(msg, "B6011",5)==0)
		{
			code_num = (char*)substring(msg,5,16); // 종목코드
			sprintf(rm_routing_message, "%s.B6011", code_num);
			amqp_basic_publish(rm_conn, 1, amqp_cstring_bytes(rm_exchange), amqp_cstring_bytes(rm_routing_message),0,0,NULL,amqp_cstring_bytes(msg));

			reply = redisCommand(redis, "SET %s %s", rm_routing_message, msg);
			freeReplyObject(reply);

			// tmp = (char*)substring(msg,29,37); // 매도호가1 4.2
			// sprintf(rm_routing_message, "호가.%s.매도호가1", code_num);
			// amqp_basic_publish(rm_conn, 1, amqp_cstring_bytes(rm_exchange), amqp_cstring_bytes(rm_routing_message),0,0,NULL,amqp_cstring_bytes(tmp));
			// free(tmp);

			// tmp = (char*)substring(msg,47,58); // 매도잔량1 4.1
			// sprintf(rm_routing_message, "호가.%s.매도잔량1", code_num);
			// amqp_basic_publish(rm_conn, 1, amqp_cstring_bytes(rm_exchange), amqp_cstring_bytes(rm_routing_message),0,0,NULL,amqp_cstring_bytes(tmp));
			// free(tmp);

			// tmp = (char*)substring(msg,38,46); // 매수호가1 5.2
			// sprintf(rm_routing_message, "호가.%s.매수호가1", code_num);
			// amqp_basic_publish(rm_conn, 1, amqp_cstring_bytes(rm_exchange), amqp_cstring_bytes(rm_routing_message),0,0,NULL,amqp_cstring_bytes(tmp));
			// free(tmp);

			// tmp = (char*)substring(msg,59,70); // 매수잔량1 5.3
			// sprintf(rm_routing_message, "호가.%s.매수잔량1", code_num);
			// amqp_basic_publish(rm_conn, 1, amqp_cstring_bytes(rm_exchange), amqp_cstring_bytes(rm_routing_message),0,0,NULL,amqp_cstring_bytes(tmp));
			// free(tmp);

			// tmp = (char*)substring(msg,71,79); // 매도호가2 3.2
			// sprintf(rm_routing_message, "호가.%s.매도호가2", code_num);
			// amqp_basic_publish(rm_conn, 1, amqp_cstring_bytes(rm_exchange), amqp_cstring_bytes(rm_routing_message),0,0,NULL,amqp_cstring_bytes(tmp));
			// free(tmp);

			// tmp = (char*)substring(msg,89,100); // 매도잔량2 3.1
			// sprintf(rm_routing_message, "호가.%s.매도잔량2", code_num);
			// amqp_basic_publish(rm_conn, 1, amqp_cstring_bytes(rm_exchange), amqp_cstring_bytes(rm_routing_message),0,0,NULL,amqp_cstring_bytes(tmp));
			// free(tmp);

			// tmp = (char*)substring(msg,80,88); // 매수호가2 6.2
			// sprintf(rm_routing_message, "호가.%s.매수호가2", code_num);
			// amqp_basic_publish(rm_conn, 1, amqp_cstring_bytes(rm_exchange), amqp_cstring_bytes(rm_routing_message),0,0,NULL,amqp_cstring_bytes(tmp));
			// free(tmp);

			// tmp = (char*)substring(msg,101,112); // 매수잔량2 6.3
			// sprintf(rm_routing_message, "호가.%s.매수잔량2", code_num);
			// amqp_basic_publish(rm_conn, 1, amqp_cstring_bytes(rm_exchange), amqp_cstring_bytes(rm_routing_message),0,0,NULL,amqp_cstring_bytes(tmp));
			// free(tmp);

			// tmp = (char*)substring(msg,113,121); // 매도호가3 2.2
			// sprintf(rm_routing_message, "호가.%s.매도호가3", code_num);
			// amqp_basic_publish(rm_conn, 1, amqp_cstring_bytes(rm_exchange), amqp_cstring_bytes(rm_routing_message),0,0,NULL,amqp_cstring_bytes(tmp));
			// free(tmp);

			// tmp = (char*)substring(msg,122,130); // 매수호가3 7.2
			// sprintf(rm_routing_message, "호가.%s.매수호가3", code_num);
			// amqp_basic_publish(rm_conn, 1, amqp_cstring_bytes(rm_exchange), amqp_cstring_bytes(rm_routing_message),0,0,NULL,amqp_cstring_bytes(tmp));
			// free(tmp);

			// tmp = (char*)substring(msg,131,142); // 매도잔량3 2.1
			// sprintf(rm_routing_message, "호가.%s.매도잔량3", code_num);
			// amqp_basic_publish(rm_conn, 1, amqp_cstring_bytes(rm_exchange), amqp_cstring_bytes(rm_routing_message),0,0,NULL,amqp_cstring_bytes(tmp));
			// free(tmp);

			// tmp = (char*)substring(msg,143,154); // 매수잔량3 7.3
			// sprintf(rm_routing_message, "호가.%s.매수잔량3", code_num);
			// amqp_basic_publish(rm_conn, 1, amqp_cstring_bytes(rm_exchange), amqp_cstring_bytes(rm_routing_message),0,0,NULL,amqp_cstring_bytes(tmp));
			// free(tmp);

			// tmp = (char*)substring(msg,155,163); // 매도호가4 1.2
			// sprintf(rm_routing_message, "호가.%s.매도호가4", code_num);
			// amqp_basic_publish(rm_conn, 1, amqp_cstring_bytes(rm_exchange), amqp_cstring_bytes(rm_routing_message),0,0,NULL,amqp_cstring_bytes(tmp));
			// free(tmp);

			// tmp = (char*)substring(msg,173,184); // 매도잔량4 1.1
			// sprintf(rm_routing_message, "호가.%s.매도잔량4", code_num);
			// amqp_basic_publish(rm_conn, 1, amqp_cstring_bytes(rm_exchange), amqp_cstring_bytes(rm_routing_message),0,0,NULL,amqp_cstring_bytes(tmp));
			// free(tmp);

			// tmp = (char*)substring(msg,164,172); // 매수호가4 8.2
			// sprintf(rm_routing_message, "호가.%s.매수호가4", code_num);
			// amqp_basic_publish(rm_conn, 1, amqp_cstring_bytes(rm_exchange), amqp_cstring_bytes(rm_routing_message),0,0,NULL,amqp_cstring_bytes(tmp));
			// free(tmp);

			// tmp = (char*)substring(msg,185,196); // 매수잔량4 8.3
			// sprintf(rm_routing_message, "호가.%s.매수잔량4", code_num);
			// amqp_basic_publish(rm_conn, 1, amqp_cstring_bytes(rm_exchange), amqp_cstring_bytes(rm_routing_message),0,0,NULL,amqp_cstring_bytes(tmp));
			// free(tmp);

			// tmp = (char*)substring(msg,197,205); // 매도호가5 0.2
			// sprintf(rm_routing_message, "호가.%s.매도호가5", code_num);
			// amqp_basic_publish(rm_conn, 1, amqp_cstring_bytes(rm_exchange), amqp_cstring_bytes(rm_routing_message),0,0,NULL,amqp_cstring_bytes(tmp));
			// free(tmp);

			// tmp = (char*)substring(msg,215,226); // 매도잔량5 0.1
			// sprintf(rm_routing_message, "호가.%s.매도잔량5", code_num);
			// amqp_basic_publish(rm_conn, 1, amqp_cstring_bytes(rm_exchange), amqp_cstring_bytes(rm_routing_message),0,0,NULL,amqp_cstring_bytes(tmp));
			// free(tmp);

			// tmp = (char*)substring(msg,206,214); // 매수호가5 9.2
			// sprintf(rm_routing_message, "호가.%s.매수호가5", code_num);
			// amqp_basic_publish(rm_conn, 1, amqp_cstring_bytes(rm_exchange), amqp_cstring_bytes(rm_routing_message),0,0,NULL,amqp_cstring_bytes(tmp));
			// free(tmp);

			// tmp = (char*)substring(msg,227,238); // 매수잔량5 9.3
			// sprintf(rm_routing_message, "호가.%s.매수잔량5", code_num);
			// amqp_basic_publish(rm_conn, 1, amqp_cstring_bytes(rm_exchange), amqp_cstring_bytes(rm_routing_message),0,0,NULL,amqp_cstring_bytes(tmp));
			// free(tmp);

			// tmp = (char*)substring(msg,449,460); // 매도총잔량 10.1
			// sprintf(rm_routing_message, "호가.%s.매도총잔량", code_num);
			// amqp_basic_publish(rm_conn, 1, amqp_cstring_bytes(rm_exchange), amqp_cstring_bytes(rm_routing_message),0,0,NULL,amqp_cstring_bytes(tmp));
			// free(tmp);

			// tmp = (char*)substring(msg,461,472); // 매수총잔량 10.3
			// sprintf(rm_routing_message, "호가.%s.매도수총잔량", code_num);
			// amqp_basic_publish(rm_conn, 1, amqp_cstring_bytes(rm_exchange), amqp_cstring_bytes(rm_routing_message),0,0,NULL,amqp_cstring_bytes(tmp));
			// free(tmp);

			free(code_num);
		}//B6011
	}
	return 0;
}
